package com.zdxk.msgPush.common.rocket.consume;

import com.zdxk.msgPush.common.rocket.consume.IMsgHandler;
import org.apache.rocketmq.client.consumer.DefaultMQPushConsumer;
import org.apache.rocketmq.client.consumer.listener.ConsumeConcurrentlyContext;
import org.apache.rocketmq.client.consumer.listener.ConsumeConcurrentlyStatus;
import org.apache.rocketmq.client.consumer.listener.MessageListenerConcurrently;
import org.apache.rocketmq.client.exception.MQClientException;
import org.apache.rocketmq.common.consumer.ConsumeFromWhere;
import org.apache.rocketmq.common.message.MessageExt;

import java.util.List;

public class RocketMqPushConsumer {

    private DefaultMQPushConsumer mqConsumer;

    private IMsgHandler imsgHandler;

    public RocketMqPushConsumer(String consumeGroupName,String nameSvrAddr,IMsgHandler imsgHandler){
        this.imsgHandler = imsgHandler; //添加消息回调处理类

        this.mqConsumer = new DefaultMQPushConsumer(consumeGroupName);
        this.mqConsumer.setNamesrvAddr(nameSvrAddr);

        // 这里设置的是一个consumer的消费策略
        // CONSUME_FROM_LAST_OFFSET 默认策略，从该队列最尾开始消费，即跳过历史消息
        // CONSUME_FROM_FIRST_OFFSET 从队列最开始开始消费，即历史消息（还储存在broker的）全部消费一遍
        // CONSUME_FROM_TIMESTAMP 从某个时间点开始消费，和setConsumeTimestamp()配合使用，默认是半个小时以前
        this.mqConsumer.setConsumeFromWhere(ConsumeFromWhere.CONSUME_FROM_FIRST_OFFSET);
        // 设置一个Listener，主要进行消息的逻辑处理
        this.mqConsumer.registerMessageListener(new MessageListenerConcurrently(){
            @Override
            public ConsumeConcurrentlyStatus consumeMessage(List<MessageExt> list, ConsumeConcurrentlyContext consumeConcurrentlyContext) {
                imsgHandler.dealMsg(list,consumeConcurrentlyContext);
                return ConsumeConcurrentlyStatus.CONSUME_SUCCESS;
            }
        });
    }

    public void subscribe(String topic, String subExpression) throws MQClientException {
        this.mqConsumer.subscribe(topic,subExpression);
    }

    public void start() throws MQClientException {
        this.mqConsumer.start();
    }
}
